{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PySpark Cookbook\n",
    "\n",
    "### Tomasz Drabas, Denny Lee\n",
    "#### Version: 0.1\n",
    "#### Date: 6/5/2018"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create some sample data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data = sc.parallelize([\n",
    "      (1, 'MacBook Pro', 2015, '15\"', '16GB', '512GB SSD', 13.75, 9.48, 0.61, 4.02)\n",
    "    , (2, 'MacBook', 2016, '12\"', '8GB', '256GB SSD', 11.04, 7.74, 0.52, 2.03)\n",
    "    , (3, 'MacBook Air', 2016, '13.3\"', '8GB', '128GB SSD', 12.8, 8.94, 0.68, 2.96)\n",
    "    , (4, 'iMac', 2017, '27\"', '64GB', '1TB SSD', 25.6, 8.0, 20.3, 20.8)\n",
    "])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating DataFrames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_df = spark.createDataFrame(\n",
    "    sample_data\n",
    "    , [\n",
    "        'Id'\n",
    "        , 'Model'\n",
    "        , 'Year'\n",
    "        , 'ScreenSize'\n",
    "        , 'RAM'\n",
    "        , 'HDD'\n",
    "        , 'W'\n",
    "        , 'D'\n",
    "        , 'H'\n",
    "        , 'Weight'\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, 'MacBook Pro', 2015, '15\"', '16GB', '512GB SSD', 13.75, 9.48, 0.61, 4.02)]"
      ]
     },
     "execution_count": 52,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Id=1, Model='MacBook Pro', Year=2015, ScreenSize='15\"', RAM='16GB', HDD='512GB SSD', W=13.75, D=9.48, H=0.61, Weight=4.02)]"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_df.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|       15\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|       12\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|     13.3\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|       27\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Id: long (nullable = true)\n",
      " |-- Model: string (nullable = true)\n",
      " |-- Year: long (nullable = true)\n",
      " |-- ScreenSize: string (nullable = true)\n",
      " |-- RAM: string (nullable = true)\n",
      " |-- HDD: string (nullable = true)\n",
      " |-- W: double (nullable = true)\n",
      " |-- D: double (nullable = true)\n",
      " |-- H: double (nullable = true)\n",
      " |-- Weight: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## From JSON"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_json_df = (\n",
    "    spark\n",
    "    .read\n",
    "    .json('../Data/DataFrames_sample.json')\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+---------+-----------+----+-----------+-----+-------+-----+---+\n",
      "|   D|   H|      HDD|      Model| RAM| ScreenSize|    W| Weight| Year| Id|\n",
      "+----+----+---------+-----------+----+-----------+-----+-------+-----+---+\n",
      "|9.48|0.61|512GB SSD|MacBook Pro|16GB|        15\"|13.75|   4.02| 2015|  1|\n",
      "|7.74|0.52|256GB SSD|    MacBook| 8GB|        12\"|11.04|   2.03| 2016|  2|\n",
      "|8.94|0.68|128GB SSD|MacBook Air| 8GB|      13.3\"| 12.8|   2.96| 2016|  3|\n",
      "| 8.0|20.3|  1TB SSD|       iMac|64GB|        27\"| 25.6|   20.8| 2017|  4|\n",
      "+----+----+---------+-----------+----+-----------+-----+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_json_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |--  D: double (nullable = true)\n",
      " |--  H: double (nullable = true)\n",
      " |--  HDD: string (nullable = true)\n",
      " |--  Model: string (nullable = true)\n",
      " |--  RAM: string (nullable = true)\n",
      " |--  ScreenSize: string (nullable = true)\n",
      " |--  W: double (nullable = true)\n",
      " |--  Weight: double (nullable = true)\n",
      " |--  Year: long (nullable = true)\n",
      " |-- Id: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_json_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## From CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_csv = (\n",
    "    spark\n",
    "    .read\n",
    "    .csv(\n",
    "        '../Data/DataFrames_sample.csv'\n",
    "        , header=True\n",
    "        , inferSchema=True)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-----+-----------+----+---------+-----+----+----+-------+\n",
      "| Id|      Model| Year| ScreenSize| RAM|      HDD|    W|   D|   H| Weight|\n",
      "+---+-----------+-----+-----------+----+---------+-----+----+----+-------+\n",
      "|  1|MacBook Pro| 2015|        15\"|16GB|512GB SSD|13.75|9.48|0.61|   4.02|\n",
      "|  2|    MacBook| 2016|        12\"| 8GB|256GB SSD|11.04|7.74|0.52|   2.03|\n",
      "|  3|MacBook Air| 2016|      13.3\"| 8GB|128GB SSD| 12.8|8.94|0.68|   2.96|\n",
      "|  4|       iMac| 2017|        27\"|64GB|  1TB SSD| 25.6| 8.0|20.3|   20.8|\n",
      "+---+-----------+-----+-----------+----+---------+-----+----+----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_csv.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Id: integer (nullable = true)\n",
      " |--  Model: string (nullable = true)\n",
      " |--  Year: integer (nullable = true)\n",
      " |--  ScreenSize: string (nullable = true)\n",
      " |--  RAM: string (nullable = true)\n",
      " |--  HDD: string (nullable = true)\n",
      " |--  W: double (nullable = true)\n",
      " |--  D: double (nullable = true)\n",
      " |--  H: double (nullable = true)\n",
      " |--  Weight: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_csv.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Accessing underlying RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Id=1, Model='MacBook Pro', Year=2015, ScreenSize='15\"', RAM='16GB', HDD='512GB SSD', W=13.75, D=9.48, H=0.61, Weight=4.02)]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_df.rdd.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(1, 'MacBook Pro', 2015, '15\"', '16GB', '512GB SSD', 13.75, 9.48, 0.61, 4.02)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+--------+--------+-----------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|HDD_size|HDD_type|Volume_cuIn|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+--------+--------+-----------+\n",
      "|  1|MacBook Pro|2015|       15\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|   512GB|     SSD|       80.0|\n",
      "|  2|    MacBook|2016|       12\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|   256GB|     SSD|       44.0|\n",
      "|  3|MacBook Air|2016|     13.3\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|   128GB|     SSD|       78.0|\n",
      "|  4|       iMac|2017|       27\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|     1TB|     SSD|     4157.0|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+--------+--------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import pyspark.sql as sql\n",
    "import pyspark.sql.functions as f\n",
    "\n",
    "sample_data_transformed = (\n",
    "    sample_data_df\n",
    "    .rdd\n",
    "    .map(lambda row: sql.Row(\n",
    "        **row.asDict()\n",
    "        , HDD_size=row.HDD.split(' ')[0]\n",
    "        )\n",
    "    )\n",
    "    .map(lambda row: sql.Row(\n",
    "        **row.asDict()\n",
    "        , HDD_type=row.HDD.split(' ')[1]\n",
    "        )\n",
    "    )\n",
    "    .map(lambda row: sql.Row(\n",
    "        **row.asDict()\n",
    "        , Volume=row.H * row.D * row.W\n",
    "        )\n",
    "    )\n",
    "    .toDF()\n",
    "    .select(\n",
    "        sample_data_df.columns + \n",
    "        [\n",
    "              'HDD_size'\n",
    "            , 'HDD_type'\n",
    "            , f.round(\n",
    "                f.col('Volume')\n",
    "            ).alias('Volume_cuIn')\n",
    "        ])\n",
    ")\n",
    "\n",
    "sample_data_transformed.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Performance optimizations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------------------+\n",
      "| id|                val|\n",
      "+---+-------------------+\n",
      "|  0| 0.9453946488613437|\n",
      "|  1|0.39388041568859766|\n",
      "|  2| 0.1356767412456391|\n",
      "+---+-------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import pyspark.sql.functions as f\n",
    "\n",
    "big_df = (\n",
    "    spark\n",
    "    .range(0, 1000000)\n",
    "    .withColumn('val', f.rand())\n",
    ")\n",
    "\n",
    "big_df.cache()\n",
    "big_df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "245 ms ± 64 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "from scipy import stats\n",
    "\n",
    "@f.pandas_udf('double', f.PandasUDFType.SCALAR)\n",
    "def pandas_pdf(v):\n",
    "    return pd.Series(stats.norm.pdf(v))\n",
    "\n",
    "def test_pandas_pdf():\n",
    "    return (big_df\n",
    "            .withColumn('probability', pandas_pdf(big_df.val))\n",
    "            .agg(f.count(f.col('probability')))\n",
    "            .show()\n",
    "        )\n",
    "\n",
    "%timeit -n 1 test_pandas_pdf()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------------------+-------------------+\n",
      "| id|                 val|        probability|\n",
      "+---+--------------------+-------------------+\n",
      "|  0|  0.9453946488613437| 0.2551703151423011|\n",
      "|  1| 0.39388041568859766| 0.3691657960230967|\n",
      "|  2|  0.1356767412456391| 0.3952872266471388|\n",
      "|  3|0.050985087503938376|0.39842409615712904|\n",
      "|  4|  0.5167556509690651|  0.349079100100191|\n",
      "+---+--------------------+-------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    big_df\n",
    "    .withColumn('probability', pandas_pdf(big_df.val))\n",
    "    .show(5)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "+------------------+\n",
      "|count(probability)|\n",
      "+------------------+\n",
      "|           1000000|\n",
      "+------------------+\n",
      "\n",
      "23.1 s ± 937 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)\n"
     ]
    }
   ],
   "source": [
    "@f.udf('double')\n",
    "def pdf(v):\n",
    "    return float(stats.norm.pdf(v))\n",
    "\n",
    "def test_pdf():\n",
    "    return (big_df\n",
    "            .withColumn('probability', pdf(big_df.val))\n",
    "            .agg(f.count(f.col('probability')))\n",
    "            .show()\n",
    "        )\n",
    "\n",
    "%timeit -n 1 test_pdf()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Infering the schema using reflection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Id, Model, Year, ScreenSize, RAM, HDD, W, D, H, Weight',\n",
       " '1,MacBook Pro,2015,\"15\\\\\"\",16GB,512GB SSD,13.75,9.48,0.61,4.02']"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_rdd = sc.textFile('../Data/DataFrames_sample.csv')\n",
    "sample_data_rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql as sql\n",
    "\n",
    "header = sample_data_rdd.first()\n",
    "\n",
    "sample_data_rdd_row = (\n",
    "    sample_data_rdd\n",
    "    .filter(lambda row: row != header)\n",
    "    .map(lambda row: row.split(','))\n",
    "    .map(lambda row:\n",
    "        sql.Row(\n",
    "            Id=int(row[0])\n",
    "            , Model=row[1]\n",
    "            , Year=int(row[2])\n",
    "            , ScreenSize=row[3]\n",
    "            , RAM=row[4]\n",
    "            , HDD=row[5]\n",
    "            , W=float(row[6])\n",
    "            , D=float(row[7])\n",
    "            , H=float(row[8])\n",
    "            , Weight=float(row[9])\n",
    "        )\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+---------+---+-----------+----+----------+-----+------+----+\n",
      "|   D|   H|      HDD| Id|      Model| RAM|ScreenSize|    W|Weight|Year|\n",
      "+----+----+---------+---+-----------+----+----------+-----+------+----+\n",
      "|9.48|0.61|512GB SSD|  1|MacBook Pro|16GB|    \"15\\\"\"|13.75|  4.02|2015|\n",
      "|7.74|0.52|256GB SSD|  2|    MacBook| 8GB|    \"12\\\"\"|11.04|  2.03|2016|\n",
      "|8.94|0.68|128GB SSD|  3|MacBook Air| 8GB|  \"13.3\\\"\"| 12.8|  2.96|2016|\n",
      "+----+----+---------+---+-----------+----+----------+-----+------+----+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.createDataFrame(sample_data_rdd_row).show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Specifying the schema programmatically"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.types as typ\n",
    "\n",
    "sch = typ.StructType([\n",
    "      typ.StructField('Id', typ.LongType(), False)\n",
    "    , typ.StructField('Model', typ.StringType(), True)\n",
    "    , typ.StructField('Year', typ.IntegerType(), True)\n",
    "    , typ.StructField('ScreenSize', typ.StringType(), True)\n",
    "    , typ.StructField('RAM', typ.StringType(), True)\n",
    "    , typ.StructField('HDD', typ.StringType(), True)\n",
    "    , typ.StructField('W', typ.DoubleType(), True)\n",
    "    , typ.StructField('D', typ.DoubleType(), True)\n",
    "    , typ.StructField('H', typ.DoubleType(), True)\n",
    "    , typ.StructField('Weight', typ.DoubleType(), True)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_rdd = sc.textFile('../Data/DataFrames_sample.csv')\n",
    "\n",
    "header = sample_data_rdd.first()\n",
    "\n",
    "sample_data_rdd = (\n",
    "    sample_data_rdd\n",
    "    .filter(lambda row: row != header)\n",
    "    .map(lambda row: row.split(','))\n",
    "    .map(lambda row: (\n",
    "                int(row[0])\n",
    "                , row[1]\n",
    "                , int(row[2])\n",
    "                , row[3]\n",
    "                , row[4]\n",
    "                , row[5]\n",
    "                , float(row[6])\n",
    "                , float(row[7])\n",
    "                , float(row[8])\n",
    "                , float(row[9])\n",
    "        )\n",
    "    )\n",
    ")\n",
    "\n",
    "sample_data_schema = spark.createDataFrame(sample_data_rdd, schema=sch)\n",
    "sample_data_schema.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating a temporary table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_schema.createTempView('sample_data_view')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----+----+---------+\n",
      "|      Model|Year| RAM|      HDD|\n",
      "+-----------+----+----+---------+\n",
      "|MacBook Pro|2015|16GB|512GB SSD|\n",
      "|    MacBook|2016| 8GB|256GB SSD|\n",
      "|MacBook Air|2016| 8GB|128GB SSD|\n",
      "|       iMac|2017|64GB|  1TB SSD|\n",
      "+-----------+----+----+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "    SELECT Model\n",
    "        , Year\n",
    "        , RAM\n",
    "        , HDD\n",
    "    FROM sample_data_view\n",
    "''').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_schema.createOrReplaceTempView('sample_data_view')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----+----+---------+----------+\n",
      "|      Model|Year| RAM|      HDD|ScreenSize|\n",
      "+-----------+----+----+---------+----------+\n",
      "|MacBook Pro|2015|16GB|512GB SSD|    \"15\\\"\"|\n",
      "|    MacBook|2016| 8GB|256GB SSD|    \"12\\\"\"|\n",
      "|MacBook Air|2016| 8GB|128GB SSD|  \"13.3\\\"\"|\n",
      "|       iMac|2017|64GB|  1TB SSD|    \"27\\\"\"|\n",
      "+-----------+----+----+---------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "    SELECT Model\n",
    "        , Year\n",
    "        , RAM\n",
    "        , HDD\n",
    "        , ScreenSize\n",
    "    FROM sample_data_view\n",
    "''').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Using SQL to interact with DataFrames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "models_df = sc.parallelize([\n",
    "      ('MacBook Pro', 'Laptop')\n",
    "    , ('MacBook', 'Laptop')\n",
    "    , ('MacBook Air', 'Laptop')\n",
    "    , ('iMac', 'Desktop')\n",
    "]).toDF(['Model', 'FormFactor'])\n",
    "\n",
    "models_df.createOrReplaceTempView('models')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_schema.createOrReplaceTempView('sample_data_view')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+----------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|FormFactor|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+----------+\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|   Desktop|\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|    Laptop|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|    Laptop|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|    Laptop|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "    SELECT a.*\n",
    "        , b.FormFactor\n",
    "    FROM sample_data_view AS a\n",
    "    LEFT JOIN models AS b\n",
    "        ON a.Model == b.Model\n",
    "    ORDER BY Weight DESC\n",
    "''').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+\n",
      "|FormFactor|ComputerCnt|\n",
      "+----------+-----------+\n",
      "|    Laptop|          3|\n",
      "|   Desktop|          1|\n",
      "+----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql('''\n",
    "    SELECT b.FormFactor\n",
    "        , COUNT(*) AS ComputerCnt\n",
    "    FROM sample_data_view AS a\n",
    "    LEFT JOIN models AS b\n",
    "        ON a.Model == b.Model\n",
    "    GROUP BY FormFactor\n",
    "''').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Overview of DataFrame transformations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.cache().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.coalesce(1).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+\n",
      "| RAM|\n",
      "+----+\n",
      "|64GB|\n",
      "|16GB|\n",
      "| 8GB|\n",
      "+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.select('RAM').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.dropDuplicates().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.dropna().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|      Model|FormFactor|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|       null|      null|\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|MacBook Pro|    Laptop|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|MacBook Air|    Laptop|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|       iMac|   Desktop|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "models_df = sc.parallelize([\n",
    "      ('MacBook Pro', 'Laptop')\n",
    "    , ('MacBook Air', 'Laptop')\n",
    "    , ('iMac', 'Desktop')\n",
    "]).toDF(['Model', 'FormFactor'])\n",
    "\n",
    "(\n",
    "    sample_data_schema\n",
    "    .join(\n",
    "        models_df\n",
    "        , sample_data_schema.Model == models_df.Model\n",
    "        , 'left'\n",
    "    ).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|      Model|FormFactor|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|MacBook Pro|    Laptop|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|MacBook Air|    Laptop|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|       iMac|   Desktop|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+-----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .join(\n",
    "        models_df\n",
    "        , sample_data_schema.Model == models_df.Model\n",
    "        , 'right'\n",
    "    ).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-------+----+----------+---+---------+-----+----+----+------+\n",
      "| Id|  Model|Year|ScreenSize|RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-------+----+----------+---+---------+-----+----+----+------+\n",
      "|  2|MacBook|2016|    \"12\\\"\"|8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "+---+-------+----+----------+---+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .join(\n",
    "        models_df\n",
    "        , sample_data_schema.Model == models_df.Model\n",
    "        , 'left_anti'\n",
    "    ).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .join(\n",
    "        models_df\n",
    "        , sample_data_schema.Model == models_df.Model\n",
    "        , 'left_semi'\n",
    "    ).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.fillna(0).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.filter(sample_data_schema.Year > 2015).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|    RAM_freqItems|\n",
      "+-----------------+\n",
      "|[64GB, 16GB, 8GB]|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.freqItems(['RAM']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.orderBy('W').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.orderBy(f.col('H').desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_schema_rep = (\n",
    "    sample_data_schema\n",
    "    .repartition(2, 'Year')\n",
    ")\n",
    "\n",
    "sample_data_schema_rep.rdd.getNumPartitions()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+\n",
      "|   A|   B|   C|\n",
      "+----+----+----+\n",
      "|21.4|36.3|24.2|\n",
      "| 1.6|32.1|27.9|\n",
      "| 3.2|38.7|24.7|\n",
      "| 2.8|21.4|23.9|\n",
      "| 3.9|34.1|27.9|\n",
      "| 9.2|21.4|21.4|\n",
      "+----+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "missing_df = sc.parallelize([\n",
    "    (None, 36.3, 24.2)\n",
    "    , (1.6, 32.1, 27.9)\n",
    "    , (3.2, 38.7, 24.7)\n",
    "    , (2.8, None, 23.9)\n",
    "    , (3.9, 34.1, 27.9)\n",
    "    , (9.2, None, None)\n",
    "]).toDF(['A', 'B', 'C'])\n",
    "\n",
    "missing_df.fillna(21.4).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+-----+\n",
      "|   A|                 B|    C|\n",
      "+----+------------------+-----+\n",
      "|4.14|              36.3| 24.2|\n",
      "| 1.6|              32.1| 27.9|\n",
      "| 3.2|              38.7| 24.7|\n",
      "| 2.8|35.300000000000004| 23.9|\n",
      "| 3.9|              34.1| 27.9|\n",
      "| 9.2|35.300000000000004|25.72|\n",
      "+----+------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "miss_dict = (\n",
    "    missing_df\n",
    "    .agg(\n",
    "        f.mean('A').alias('A')\n",
    "        , f.mean('B').alias('B')\n",
    "        , f.mean('C').alias('C')\n",
    "    )\n",
    ").toPandas().to_dict('records')[0]\n",
    "\n",
    "missing_df.fillna(miss_dict).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+----+\n",
      "|  A|   B|   C|\n",
      "+---+----+----+\n",
      "|1.6|32.1|27.9|\n",
      "|3.2|38.7|24.7|\n",
      "|3.9|34.1|27.9|\n",
      "+---+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "missing_df.dropna().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----+----+\n",
      "|   A|   B|   C|\n",
      "+----+----+----+\n",
      "|null|36.3|24.2|\n",
      "| 1.6|32.1|27.9|\n",
      "| 3.2|38.7|24.7|\n",
      "| 2.8|null|23.9|\n",
      "| 3.9|34.1|27.9|\n",
      "+----+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "missing_df.dropna(thresh=2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 135,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----+----+\n",
      "|  A|   B|   C|\n",
      "+---+----+----+\n",
      "|1.6|32.1|27.9|\n",
      "|3.9|34.1|27.9|\n",
      "|3.2|38.7|24.7|\n",
      "+---+----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dupes_df = sc.parallelize([\n",
    "      (1.6, 32.1, 27.9)\n",
    "    , (3.2, 38.7, 24.7)\n",
    "    , (3.9, 34.1, 27.9)\n",
    "    , (3.2, 38.7, 24.7)\n",
    "]).toDF(['A', 'B', 'C'])\n",
    "\n",
    "dupes_df.dropDuplicates().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+\n",
      "|      Model|ScreenSize|\n",
      "+-----------+----------+\n",
      "|MacBook Pro|    \"15\\\"\"|\n",
      "|    MacBook|    \"12\\\"\"|\n",
      "|MacBook Air|  \"13.3\\\"\"|\n",
      "|       iMac|    \"27\\\"\"|\n",
      "+-----------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.select('Model', 'ScreenSize').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|                 W|\n",
      "+-------+------------------+\n",
      "|  count|                 4|\n",
      "|   mean|15.797500000000001|\n",
      "| stddev| 6.630738395281983|\n",
      "|    min|             11.04|\n",
      "|    25%|             11.04|\n",
      "|    50%|              12.8|\n",
      "|    75%|             13.75|\n",
      "|    max|              25.6|\n",
      "+-------+------------------+\n",
      "\n",
      "+-------+------------------+\n",
      "|summary|                 W|\n",
      "+-------+------------------+\n",
      "|  count|                 4|\n",
      "|   mean|15.797500000000001|\n",
      "| stddev| 6.630738395281983|\n",
      "|    min|             11.04|\n",
      "|    max|              25.6|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.select('W').summary().show()\n",
    "sample_data_schema.select('W').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|\n",
      "|  5|MacBook Pro|2018|       15\"|16GB|256GB SSD|13.75|9.48|0.61|  4.02|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "another_macBookPro = sc.parallelize([\n",
    "      (5, 'MacBook Pro', 2018, '15\"', '16GB', '256GB SSD', 13.75, 9.48, 0.61, 4.02)\n",
    "]).toDF(sample_data_schema.columns)\n",
    "\n",
    "sample_data_schema.unionAll(another_macBookPro).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 136,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|    HDDSplit|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|[512GB, SSD]|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|[256GB, SSD]|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|[128GB, SSD]|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|  [1TB, SSD]|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .withColumn('HDDSplit', f.split(f.col('HDD'), ' '))\n",
    "    .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 140,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "| RAM|count|\n",
      "+----+-----+\n",
      "|64GB|    1|\n",
      "|16GB|    1|\n",
      "| 8GB|    2|\n",
      "+----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .groupBy('RAM')\n",
    "    .count()\n",
    "    .show()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "| Id|      Model|Year|ScreenSize| RAM|      HDD|    W|   D|   H|Weight|      Volume|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "|  1|MacBook Pro|2015|    \"15\\\"\"|16GB|512GB SSD|13.75|9.48|0.61|  4.02|[512GB, SSD]|\n",
      "|  2|    MacBook|2016|    \"12\\\"\"| 8GB|256GB SSD|11.04|7.74|0.52|  2.03|[256GB, SSD]|\n",
      "|  3|MacBook Air|2016|  \"13.3\\\"\"| 8GB|128GB SSD| 12.8|8.94|0.68|  2.96|[128GB, SSD]|\n",
      "|  4|       iMac|2017|    \"27\\\"\"|64GB|  1TB SSD| 25.6| 8.0|20.3|  20.8|  [1TB, SSD]|\n",
      "+---+-----------+----+----------+----+---------+-----+----+----+------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    sample_data_schema\n",
    "    .select(\n",
    "        f.col('*')\n",
    "        , f.split(f.col('HDD'), ' ').alias('HDD_Array')\n",
    "    ).show()\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Overview of DataFrame actions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Year=2015, count=1), Row(Year=2016, count=2), Row(Year=2017, count=1)]"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_schema.groupBy('Year').count().collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|                 W|\n",
      "+-------+------------------+\n",
      "|  count|                 4|\n",
      "|   mean|15.797500000000001|\n",
      "| stddev| 6.630738395281983|\n",
      "|    min|             11.04|\n",
      "|    max|              25.6|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sample_data_schema.select('W').describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Id=1, Model='MacBook Pro', Year=2015, ScreenSize='\"15\\\\\"\"', RAM='16GB', HDD='512GB SSD', W=13.75, D=9.48, H=0.61, Weight=4.02),\n",
       " Row(Id=2, Model='MacBook', Year=2016, ScreenSize='\"12\\\\\"\"', RAM='8GB', HDD='256GB SSD', W=11.04, D=7.74, H=0.52, Weight=2.03)]"
      ]
     },
     "execution_count": 49,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_schema.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Id</th>\n",
       "      <th>Model</th>\n",
       "      <th>Year</th>\n",
       "      <th>ScreenSize</th>\n",
       "      <th>RAM</th>\n",
       "      <th>HDD</th>\n",
       "      <th>W</th>\n",
       "      <th>D</th>\n",
       "      <th>H</th>\n",
       "      <th>Weight</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>1</td>\n",
       "      <td>MacBook Pro</td>\n",
       "      <td>2015</td>\n",
       "      <td>\"15\\\"\"</td>\n",
       "      <td>16GB</td>\n",
       "      <td>512GB SSD</td>\n",
       "      <td>13.75</td>\n",
       "      <td>9.48</td>\n",
       "      <td>0.61</td>\n",
       "      <td>4.02</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2</td>\n",
       "      <td>MacBook</td>\n",
       "      <td>2016</td>\n",
       "      <td>\"12\\\"\"</td>\n",
       "      <td>8GB</td>\n",
       "      <td>256GB SSD</td>\n",
       "      <td>11.04</td>\n",
       "      <td>7.74</td>\n",
       "      <td>0.52</td>\n",
       "      <td>2.03</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>3</td>\n",
       "      <td>MacBook Air</td>\n",
       "      <td>2016</td>\n",
       "      <td>\"13.3\\\"\"</td>\n",
       "      <td>8GB</td>\n",
       "      <td>128GB SSD</td>\n",
       "      <td>12.80</td>\n",
       "      <td>8.94</td>\n",
       "      <td>0.68</td>\n",
       "      <td>2.96</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>4</td>\n",
       "      <td>iMac</td>\n",
       "      <td>2017</td>\n",
       "      <td>\"27\\\"\"</td>\n",
       "      <td>64GB</td>\n",
       "      <td>1TB SSD</td>\n",
       "      <td>25.60</td>\n",
       "      <td>8.00</td>\n",
       "      <td>20.30</td>\n",
       "      <td>20.80</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   Id        Model  Year ScreenSize   RAM        HDD      W     D      H  \\\n",
       "0   1  MacBook Pro  2015     \"15\\\"\"  16GB  512GB SSD  13.75  9.48   0.61   \n",
       "1   2      MacBook  2016     \"12\\\"\"   8GB  256GB SSD  11.04  7.74   0.52   \n",
       "2   3  MacBook Air  2016   \"13.3\\\"\"   8GB  128GB SSD  12.80  8.94   0.68   \n",
       "3   4         iMac  2017     \"27\\\"\"  64GB    1TB SSD  25.60  8.00  20.30   \n",
       "\n",
       "   Weight  \n",
       "0    4.02  \n",
       "1    2.03  \n",
       "2    2.96  \n",
       "3   20.80  "
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sample_data_schema.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_data_schema.write.mode('overwrite').csv('sample_data_schema.csv')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
